package org.example.flume.interceptor;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.List;
import java.util.Map;

/**
 * static class Builder
 * jar put it into flume-1.9.0/lib
 *
 * a1.sources.r1.selector.type = multiplexing
 * a1.sources.r1.selector.header = title
 * a1.sources.r1.selector.mapping.gao = c1
 * a1.sources.r1.selector.mapping.qiang = c2
 *
 * a1.sources.r1.interceptors = i1
 * a1.sources.r1.interceptors.i1.type = org.example.flume.interceptor.MyInterceptor$Builder
 */
public class MyInterceptor implements Interceptor {
    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {
        String body = new String(event.getBody());
        Map<String, String> headers = event.getHeaders();
        if (body.contains("gaoqiang")) {
            headers.put("title", "gao");
        } else {
            headers.put("title", "qiang");
        }
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> list) {
        for (Event event : list) {
            intercept(event);
        }
        return list;
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder {

        @Override
        public Interceptor build() {
            return new MyInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }
}
